package com.atguigu.gmall.realtime.dwd.db.app;

import com.atguigu.gmall.realtime.common.base.BaseSQLApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author atguigu
 * @Date 2023/7/7 14:16
 */
public class Dwd_05_DwdTradeOrderCancel extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_05_DwdTradeOrderCancel().start(
            30005,
            2,
            "Dwd_05_DwdTradeOrderCancel"
        );
        
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));
        
        // 1. 读取 ods_db 数据
        readOdsDb(tEnv);
        // 2. 读取 dwd 层下单事务事实表数据
        tEnv.executeSql(
            "create table dwd_trade_order_detail(" +
                "id string," +
                "order_id string," +
                "user_id string," +
                "sku_id string," +
                "sku_name string," +
                "province_id string," +
                "activity_id string," +
                "activity_rule_id string," +
                "coupon_id string," +
                "date_id string," +
                "create_time string," +
                "sku_num string," +
                "split_original_amount string," +
                "split_activity_amount string," +
                "split_coupon_amount string," +
                "split_total_amount string," +
                "ts bigint " +
                ")" + SQLUtil.getKafkaDDLSource("dwd_trade_order_detail", Constant.TOPIC_DWD_TRADE_ORDER_DETAIL));
        
        // 3. 从 ods_db 过滤出订单取消数据
        Table orderCancel = tEnv.sqlQuery("select " +
                                              " `data`['id'] id, " +
                                              " `data`['operate_time'] operate_time, " +
                                              " `ts` " +
                                              "from ods_db " +
                                              "where `database`='gmall2023' " +
                                              "and `table`='order_info' " +
                                              "and `type`='update' " +
                                              "and `old`['order_status']='1001' " +
                                              "and `data`['order_status']='1003' ");
        tEnv.createTemporaryView("order_cancel", orderCancel);
        
        
        // 4. 订单取消表和下单表进行 join
        Table result = tEnv.sqlQuery(
            "select  " +
                "od.id," +
                "od.order_id," +
                "od.user_id," +
                "od.sku_id," +
                "od.sku_name," +
                "od.province_id," +
                "od.activity_id," +
                "od.activity_rule_id," +
                "od.coupon_id," +
                "date_format(oc.operate_time, 'yyyy-MM-dd') order_cancel_date_id," +
                "oc.operate_time," +
                "od.sku_num," +
                "od.split_original_amount," +
                "od.split_activity_amount," +
                "od.split_coupon_amount," +
                "od.split_total_amount," +
                "oc.ts " +
                "from dwd_trade_order_detail od " +
                "join order_cancel oc " +
                "on od.order_id=oc.id ");
        // 5. 写出
        tEnv.executeSql(
            "create table dwd_trade_order_cancel(" +
                "id string," +
                "order_id string," +
                "user_id string," +
                "sku_id string," +
                "sku_name string," +
                "province_id string," +
                "activity_id string," +
                "activity_rule_id string," +
                "coupon_id string," +
                "date_id string," +
                "cancel_time string," +
                "sku_num string," +
                "split_original_amount string," +
                "split_activity_amount string," +
                "split_coupon_amount string," +
                "split_total_amount string," +
                "ts bigint " +
                ")" + SQLUtil.getKafkaDDLSink(Constant.TOPIC_DWD_TRADE_ORDER_CANCEL));
    
        result.executeInsert("dwd_trade_order_cancel");
    }
}
/*
交易域取消订单事务事实表
    业务过程: 取消订单  手动取消 自动取消
    
    粒度: 商品
    
    业务表:
        order_info: update and order_status 1001=>1003
        order_detail: insert   ttl:30m+5s
        order_detail_activity: insert
        order_detail_coupon: insert
        
    精简:
       order_info: update and order_status 1001=>1003
       join
       dwd 层的下单事务事实表
 */